package com.tyylab.message.core.config;

import com.tyylab.message.core.config.properties.TyylabMessageRedisProperties;
import io.lettuce.core.ClientOptions;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.TimeoutOptions;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.protocol.CommandArgsAccessor;
import io.lettuce.core.protocol.CommandType;
import io.lettuce.core.protocol.RedisCommand;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.TimeUnit;

@Configuration
public class RedisConfig {
    @Autowired
    TyylabMessageRedisProperties tyylabMessageRedisProperties;

    /**
     * 普通数据读取连接
     * @return
     */
    @Bean(name = "readConn")
    public StatefulRedisConnection<String,String> getReadStatefulConnection(){
        RedisURI redisUri = RedisURI.builder()                    // <1> 创建单机连接的连接信息
                .withHost(tyylabMessageRedisProperties.getHost())
                .withPort(tyylabMessageRedisProperties.getPort())
                .withTimeout(Duration.of(tyylabMessageRedisProperties.getTimeout(), ChronoUnit.SECONDS))
                .withPassword(tyylabMessageRedisProperties.getPassword())
                .withDatabase(tyylabMessageRedisProperties.getDatabase())
                .build();
        RedisClient redisClient = RedisClient.create(redisUri);   // <2> 创建客户端
        StatefulRedisConnection<String, String> connection = redisClient.connect();     // <3> 创建线程安全的连接
        return connection;
    }

    /**
     * 通道专用数据读取连接，会阻塞进程
     * @return
     */
    @Bean(name = "readBusConn")
    public StatefulRedisConnection<String,String> getStatefulConnection(){

        TimeoutOptions timeoutOptions = TimeoutOptions.builder().timeoutSource(new TimeoutOptions.TimeoutSource() {
            @Override
            public long getTimeout(RedisCommand<?, ?, ?> command) {

                if (command.getType() == CommandType.BLPOP
                    || command.getType() == CommandType.BRPOP
                ) {
                    return TimeUnit.MILLISECONDS.toNanos(CommandArgsAccessor.getFirstInteger(command.getArgs()));
                }
                // -1 indicates fallback to the default timeout
                return -1;
            }
        }).build();

        ClientOptions clientOptions = ClientOptions.builder().timeoutOptions(timeoutOptions).build();


        RedisURI redisUri = RedisURI.builder()                    // <1> 创建单机连接的连接信息
                .withHost(tyylabMessageRedisProperties.getHost())
                .withPort(tyylabMessageRedisProperties.getPort())
                .withTimeout(Duration.of(tyylabMessageRedisProperties.getTimeout(), ChronoUnit.SECONDS))
                .withPassword(tyylabMessageRedisProperties.getPassword())
                .withDatabase(tyylabMessageRedisProperties.getDatabase())
                .build();
        RedisClient redisClient = RedisClient.create(redisUri);   // <2> 创建客户端

        redisClient.setOptions(clientOptions);

        StatefulRedisConnection<String, String> connection = redisClient.connect();     // <3> 创建线程安全的连接

        return connection;
    }

    /**
     * 普通数据写入连接
     * @return
     */
    @Bean(name = "writeConn")
    public StatefulRedisConnection<String,String> getAsyncPublishStatefulConnection(){
        RedisURI redisUri = RedisURI.builder()                    // <1> 创建单机连接的连接信息
                .withHost(tyylabMessageRedisProperties.getHost())
                .withPort(tyylabMessageRedisProperties.getPort())
                .withTimeout(Duration.of(tyylabMessageRedisProperties.getTimeout(), ChronoUnit.SECONDS))
                .withPassword(tyylabMessageRedisProperties.getPassword())
                .withDatabase(tyylabMessageRedisProperties.getDatabase())
                .build();
        RedisClient redisClient = RedisClient.create(redisUri);   // <2> 创建客户端
        StatefulRedisConnection<String, String> connection = redisClient.connect();     // <3> 创建线程安全的连接
        return connection;
    }

}

